package org.zachary.imitation.raft.core;

import com.google.protobuf.ByteString;
import com.google.protobuf.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zachary.imitation.raft.NodeImpl;
import org.zachary.imitation.raft.Status;
import org.zachary.imitation.raft.error.RaftError;
import org.zachary.imitation.raft.option.RaftOptions;
import org.zachary.imitation.raft.option.ReplicatorOptions;
import org.zachary.imitation.raft.rpc.RpcRequests;
import org.zachary.imitation.raft.rpc.RpcResponseClosure;
import org.zachary.imitation.raft.rpc.RpcResponseClosureAdapter;
import org.zachary.imitation.raft.util.ThreadId;

import javax.annotation.concurrent.ThreadSafe;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

/**
 * 整个jraft框架中第二核心类，该类的对象就是一个日志复制器，如果当前节点是 Leader，那么集群中的其它节点都会被 Leader 包装成一个个的复制器对象，心跳检测，日志传输都是通过复制器对象进行的
 */
@ThreadSafe
public class Replicator implements ThreadId.OnError {
	private static final Logger logger = LoggerFactory.getLogger(Replicator.class);

	/** RPC 客户端 */
	// private final RaftClientService rpcService;
	/** ThreadId 可以看成一个句柄，Replicator 对象的引用会被 ThreadId 对象持用 */
	protected ThreadId id;
	/** 复制器的配置参数 */
	private final ReplicatorOptions options;
	/** 全局的配置参数 */
	private final RaftOptions raftOptions;
	/** 基于 JDK 的定时任务管理器 */
	private final ScheduledExecutorService scheduledExecutorService;

	/** Leader 最后一次发送信息（包括心跳和日志）的时间 */
	private volatile long lastRpcSendTimestamp;
	/** 心跳计数器，代表发送了多少次心跳请求 */
	private volatile long heartbeatCounter = 0;

	public Replicator(final ReplicatorOptions replicatorOptions, final RaftOptions raftOptions) {
		this.options = replicatorOptions;
		this.scheduledExecutorService = replicatorOptions.getScheduledExecutorService();
		this.raftOptions = raftOptions;
		// this.rpcService = replicatorOptions.getRaftRpcService();
	}


	/** Replicator 类的静态的核心方法，在该方法中创建了 Replicator 复制器对象 */
	public static ThreadId start(final ReplicatorOptions opts, final RaftOptions raftOptions) {
		// 创建 Replicator 复制器对象，此时的两个配置参数对象就派上用场了
		final Replicator replicator = new Replicator(opts, raftOptions);
		// 在这里远程连接这个复制器对象对应的节点，连接失败返回 null
//		if (!r.rpcService.connect(opts.getPeerId().getEndpoint())) {
//			LOG.error("Fail to init sending channel to {}.", opts.getPeerId());
//			return null;
//		}

		// 创建 ThreadId 句柄对象，这个对象持有了刚才创建的 Replicator 复制器对象
		replicator.id = new ThreadId(replicator, replicator);
		// 给当前 Replicator 复制器对象上一个锁，上锁期间，其它任何线程都不能操纵当前的复制器对象。上锁和释放锁的操作在复制器对象中非常常见，比如说发送心跳，肯定不能多个线程都在操纵这个复制器对象，来向集群中对应的节点发送心跳消息。这里上锁之后，会在发送完探针消息之后把锁释放了
		replicator.id.lock();
		logger.info("Replicator={}@{} is started", replicator.id, replicator.options.getPeerId());

		// 记录最后一次发送信息的时间
		replicator.lastRpcSendTimestamp = TimeUnit.NANOSECONDS.toMillis(System.nanoTime());

		// 启动心跳定时器，这个定时器会定时向其它节点发送心跳消息
		replicator.startHeartbeatTimer(TimeUnit.NANOSECONDS.toMillis(System.nanoTime()));

		// 发送探针请求，探针请求是 Leader 刚刚当选之后就要向集群中其它节点发送的，这个探针请求的作用就是通知其它节点，Leader 已经选举出来了，其它节点会更新它们的 Leader 信息
		// 其它节点在回复这个探针的响应时，还会把自己的日志记录告诉 Leader，让 Leader 决定传输日志的方式，看是正常传输还是使用快照
		replicator.sendProbeRequest();

		// 返回 ThreadId 对象
		return replicator.id;
	}

	/** 启动心跳定时器*/
	private void startHeartbeatTimer(final long startMs) {
		final long dueTime = startMs + this.options.getDynamicHeartBeatTimeoutMs();
		try {
			// 使用全局定时任务管理器提交一个定时任务，在这个任务中执行发送心跳消息的请求
			scheduledExecutorService.schedule(() -> onTimeout(this.id), dueTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
		} catch (final Exception e) {
			logger.error("Fail to schedule heartbeat timer", e);
			onTimeout(this.id);
		}
	}

	/** 发送心跳消息的方法，在该方法内部会调用 setError 方法，在 setError 方法内部会调用发送心跳消息的方法（{@link #onError(ThreadId, Object, int)}） */
	private static void onTimeout(final ThreadId id) {
		if (id != null) {
			id.setError(RaftError.ETIMEDOUT.getNumber());
		} else {
			logger.warn("Replicator id is null when timeout, maybe it's destroyed.");
		}
	}

	private void sendProbeRequest() {
		sendEmptyEntries(false);
	}

	/** 发送纯粹空的心跳包，不包含任何日志数据 */
	private void sendEmptyEntries(final boolean isHeartbeat) {
		sendEmptyEntries(isHeartbeat, null);
	}

	/** 发送心跳消息 */
	private static void sendHeartbeat(final ThreadId id) {
		// 发送之前先上锁
		final Replicator r = (Replicator) id.lock();
		if (r == null) {
			return;
		}
		r.sendEmptyEntries(true);
	}

	/** Leader 收到 Follower 的心跳响应的处理逻辑 */
	static void onHeartbeatReturned(final ThreadId id, final Status status, final RpcRequests.AppendEntriesRequest request, final RpcRequests.AppendEntriesResponse response, final long rpcSendTime) {
		//如果ThreadId为null，这个threadId对象可能已经被销毁了
		if (id == null) {
			return;
		}
		final long startTimeMs = System.currentTimeMillis();
		Replicator r;
		//先上锁
		if ((r = (Replicator) id.lock()) == null) {
			return;
		}
		boolean doUnlock = true;
		try {//判断是否启动了debug日志模式
			final boolean isLogDebugEnabled = logger.isDebugEnabled();
			StringBuilder sb = null;
			if (isLogDebugEnabled) {
				sb = new StringBuilder("Node ")
						.append(r.options.getGroupId())
						.append(':')
						.append(r.options.getServerId())
						.append(" received HeartbeatResponse from ")
						.append(r.options.getPeerId())
						.append(" prevLogIndex=")
						.append(request.getPrevLogIndex())
						.append(" prevLogTerm=")
						.append(request.getPrevLogTerm());
			}
			// Follower 的心跳响应中的任期更大
			if (response.getTerm() > r.options.getTerm()) {
				if (isLogDebugEnabled) {
					sb.append(" fail, greater term ")
							.append(response.getTerm())
							.append(" expect term ")
							.append(r.options.getTerm());
					logger.debug(sb.toString());
				}
				final NodeImpl node = r.options.getNode();
				// TODO 处理脑裂问题
				return;
			}

			// 响应状态为不成功
			if (!response.getSuccess() && response.hasLastLogIndex()) {
				if (isLogDebugEnabled) {
					sb.append(" fail, response term ")
							.append(response.getTerm())
							.append(" lastLogIndex ")
							.append(response.getLastLogIndex());
					logger.debug(sb.toString());
				}
				logger.warn("Heartbeat to peer {} failure, try to send a probe request.", r.options.getPeerId());
				doUnlock = false;
				// 发送探针消息
				r.sendProbeRequest();
				// 启动心跳定时器
				r.startHeartbeatTimer(startTimeMs);
				return;
			}

			if (isLogDebugEnabled) {
				logger.debug(sb.toString());
			}
			if (rpcSendTime > r.lastRpcSendTimestamp) {
				r.lastRpcSendTimestamp = rpcSendTime;
			}//走到这里不管前面结果怎样，都会启动心跳定时器
			//在这个方法中又会重新提交一个心跳定时任务给调度器
			//这里大家应该就能意识到了，领导者的心跳消息就是这样发送的，每一次收到心跳之后
			//在处理心跳的响应时，开启下一次心跳定时任务
			r.startHeartbeatTimer(startTimeMs);
		} finally {
			if (doUnlock) {
				//在这里把锁释放了
				id.unlock();
			}
		}
	}


	/** 填充 AppendEntriesRequest 请求中公共字段 */
	private boolean fillCommonFields(final RpcRequests.AppendEntriesRequest.Builder rb, long prevLogIndex, final boolean isHeartbeat) {
		rb.setTerm(this.options.getTerm());
		rb.setGroupId(this.options.getGroupId());
		rb.setServerId(this.options.getServerId().toString()); // Leader
		rb.setPeerId(this.options.getPeerId().toString()); // Follower
		rb.setPrevLogIndex(prevLogIndex);
		rb.setPrevLogTerm(0);
		rb.setCommittedIndex(0);
		return true;
	}

	/** 送心跳或者探针请求的方法，根据方法中的第一个参数判断是否发送心跳还是探针消息，第二个是收到响应后的回调方法 */
	private void sendEmptyEntries(final boolean isHeartbeat, final RpcResponseClosure<RpcRequests.AppendEntriesResponse> heartBeatClosure) {
		// 开始构建 AppendEntriesRequest 请求，该请求是共用的，日志传输、心跳消息、探针消息都会使用这个请求封装消息内容
		final RpcRequests.AppendEntriesRequest.Builder rb = RpcRequests.AppendEntriesRequest.newBuilder();
		// 填充请求中的公共属性，这里的 preLogIndex = 0 是写死的，后续版本会把这里重构成和源码一致
		fillCommonFields(rb, 0, isHeartbeat);
		try {
			// 当前发送请求的时间
			final long monotonicSendTimeMs = System.currentTimeMillis();
			// 根据方法中的参数判断是否为心跳请求
			if (isHeartbeat) {
				logger.info("Leader triggers heartbeat.");
				// 创建 AppendEntriesRequest 对象封装消息内容
				final RpcRequests.AppendEntriesRequest request = rb.build();
				// 心跳计数器自增
				this.heartbeatCounter++;

				// 定义一个回调方法，该方法会在接收到 Follower 回复的心跳响应后被回调
				RpcResponseClosure<RpcRequests.AppendEntriesResponse> heartbeatDone;
				// 判断参数中的回调方法是否为空
				if (heartBeatClosure != null) {
					heartbeatDone = heartBeatClosure;
				} else { // 如果为空就创建一个默认的回调
					heartbeatDone = new RpcResponseClosureAdapter<RpcRequests.AppendEntriesResponse>() {
						@Override
						public void run(final Status status) {
						// 在创建的新的回调方法中，收到心跳响应之后，会回调这个方法，在这个方法中会提交下一个心跳消息任务给全局定时器
						onHeartbeatReturned(Replicator.this.id, status, request, getResponse(), monotonicSendTimeMs);
						}
					};
				}

				// 发送心跳消息
				// this.rpcService.appendEntries(this.options.getPeerId().getEndpoint(), request, this.options.getElectionTimeoutMs() / 2, heartbeatDone);
			} else {
				// 探针消息
				rb.setData(ByteString.EMPTY);
				// 创建 AppendEntriesRequest 对象封装探针消息
				final RpcRequests.AppendEntriesRequest request = rb.build();
				// 发送探针消息，暂时省略回调方法
//				final Future<Message> rpcFuture = this.rpcService.appendEntries(this.options.getPeerId().getEndpoint(), request, -1, new RpcResponseClosureAdapter<RpcRequests.AppendEntriesResponse>() {
//							@Override
//							public void run(final Status status) {
//								// 回调方法暂时不做处理
//							}
//						});
			}

			logger.debug("Node {} send HeartbeatRequest to {} term {} lastCommittedIndex {}", this.options.getNode().getNodeId(), this.options.getPeerId(), this.options.getTerm(), rb.getCommittedIndex());
		} finally {
			//释放锁，这里释放的是sendHeartbeat(final ThreadId id)方法中获得的锁，就是这个方法
			//private static void sendHeartbeat(final ThreadId id) {
			//        //发送之前先上锁
			//        final Replicator r = (Replicator) id.lock();
			//        if (r == null) {
			//            return;
			//        }
			//        r.sendEmptyEntries(true);
			//    }
			unlockId();
		}
	}

	//释放锁的方法
	private void unlockId() {
		if (this.id == null) {
			return;
		}
		this.id.unlock();
	}

	/*** 该方法执行了心跳定时任务 */
	@Override
	public void onError(final ThreadId id, final Object data, final int errorCode) {
		final Replicator r = (Replicator) data;
		// RpcUtils.runInThread(() -> sendHeartbeat(id));
		sendHeartbeat(id);
	}
}
